package io.gitee.skyarthur1987.leporidae.delayqueue.core;

import com.google.common.collect.Lists;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import io.gitee.skyarthur1987.leporidae.delayqueue.autoconfigure.DelayQueuePollSchedulerConfig;
import io.gitee.skyarthur1987.leporidae.delayqueue.dto.BatchDelayQueueTask;
import io.gitee.skyarthur1987.leporidae.delayqueue.exception.DelayQueueConsumerExceptionHandler;
import io.gitee.skyarthur1987.leporidae.utils.base.JsonUtils;
import io.gitee.skyarthur1987.leporidae.utils.base.ListUtils;
import jakarta.annotation.PostConstruct;
import lombok.extern.slf4j.Slf4j;
import org.springframework.util.Assert;
import org.springframework.util.CollectionUtils;

import java.util.List;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;

/**
 * 延迟队列定时拉取
 * Created by skyarthur on 2023/1/27
 */
@Slf4j
public class DelayQueuePollScheduler {
    private final Consumer<BatchDelayQueueTask> delayQueueConsumer;

    private List<ScheduledExecutorService> scheduledExecutorServices;
    private final List<String> delayQueueTopics;

    private final DelayQueuePollSchedulerConfig schedulerConfig;

    private final DelayQueueConsumerExceptionHandler exceptionHandler;

    public DelayQueuePollScheduler(Consumer<BatchDelayQueueTask> delayQueueConsumer,
                                   DelayQueuePollSchedulerConfig schedulerConfig,
                                   List<String> delayQueueTopics,
                                   DelayQueueConsumerExceptionHandler exceptionHandler) {
        this.delayQueueConsumer = delayQueueConsumer;
        this.schedulerConfig = schedulerConfig;
        this.delayQueueTopics = delayQueueTopics;
        this.exceptionHandler = exceptionHandler;
    }

    @PostConstruct
    public void init() {
        if (CollectionUtils.isEmpty(delayQueueTopics)) {
            return;
        }
        Assert.notNull(schedulerConfig, "DelayQueuePollSchedulerConfig cannot be null!");
        int schedulerSize = schedulerConfig.getSize();
        Assert.isTrue(schedulerSize > 0, "scheduler size must > 0");
        List<List<String>> topicGroupList = ListUtils.partitionByFixedGroup(delayQueueTopics, schedulerSize);
        createExecutors();
        int i = 0;
        for (List<String> topicGroup : topicGroupList) {
            ScheduledExecutorService executorService = scheduledExecutorServices.get(i);
            log.info("ScheduledExecutorService: {}, topic list: {}", i, JsonUtils.toJson(topicGroup));
            Runnable command = () -> topicGroup.forEach(t -> {
                BatchDelayQueueTask task =
                        BatchDelayQueueTask.builder().topic(t).expireTime(System.currentTimeMillis()).build();
                try {
                    delayQueueConsumer.accept(task);
                } catch (Exception e) {
                    exceptionHandler.handle(e, task);
                }
            });
            executorService.scheduleAtFixedRate(command, schedulerConfig.getInitialDelayInMillis(),
                    schedulerConfig.getPeriodInMillis(), TimeUnit.MILLISECONDS);
            i++;
        }
    }

    /**
     * 创建定时调度 ScheduledExecutorService
     */
    private void createExecutors() {
        scheduledExecutorServices = Lists.newArrayList();
        for (int i = 0; i < schedulerConfig.getSize(); i++) {
            String threadName = "DelayQueuePollExecutor" + i + "-%d";
            scheduledExecutorServices.add(new ScheduledThreadPoolExecutor(1,
                    new ThreadFactoryBuilder().setNameFormat(threadName).setDaemon(true).build(),
                    new ThreadPoolExecutor.DiscardPolicy()));
        }
        log.info("Create scheduledExecutorServices size: {}", scheduledExecutorServices.size());
    }
}
